package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class Demo3MysqlSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //使用自定义source
        DataStreamSource<String> mysqlDS = env.addSource(new MysqlSource());

        mysqlDS.print();

        env.execute();
    }
}

/**
 * 自定义source ,实现SourceFunction接口，再run方法中编写读取数据的逻辑
 */
class MysqlSource implements SourceFunction<String> {
    /**
     * run方法用于读取外部数据源的数据方法，只会执行一次
     *
     * @param ctx :用于将读取到的数据发送到下游
     */
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        //使用JDBC读取mysql .将读取到的数据发送到下游
        //1、加载驱动
        Class.forName("com.mysql.jdbc.Driver");
        //2、创建数据库连接
        Connection con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata27", "root", "123456");
        //编写sql
        PreparedStatement stat = con.prepareStatement("select * from students");
        //执行查询
        ResultSet resultSet = stat.executeQuery();
        //循环解析数据
        while (resultSet.next()) {
            //通过字段名获取数据
            String id = resultSet.getString("id");
            String name = resultSet.getString("name");
            int age = resultSet.getInt("age");
            String sex = resultSet.getString("sex");
            String clazz = resultSet.getString("clazz");
            //将数据发送到下游
            ctx.collect(id + "," + name + "," + age + "," + sex + "," + clazz);
        }
    }

    //任务被取消时执行
    @Override
    public void cancel() {
    }
}

